package mayor.mq.producer.transaction;

import com.alibaba.rocketmq.client.producer.LocalTransactionState;
import com.alibaba.rocketmq.client.producer.SendResult;
import com.alibaba.rocketmq.client.producer.TransactionCheckListener;
import com.alibaba.rocketmq.client.producer.TransactionMQProducer;
import com.alibaba.rocketmq.common.message.Message;
import com.alibaba.rocketmq.common.message.MessageExt;

import java.util.concurrent.TimeUnit;

public class Producer {

    public static void main(String[] args) throws Exception{
        String group_name = "transaction_producer";

        final TransactionMQProducer producer = new TransactionMQProducer(group_name);
        producer.setNamesrvAddr("192.168.2.246:9876;192.168.2.190:9876");
        producer.setCheckThreadPoolMinSize(5);
        producer.setCheckThreadPoolMaxSize(20);
        //队列数
        producer.setCheckRequestHoldMax(2000);
        producer.start();
        //服务器回调，检查本地事务分支是成功还是失败
        producer.setTransactionCheckListener(new TransactionCheckListener() {
            public LocalTransactionState checkLocalTransactionState(MessageExt msg) {
                System.out.println("state-- " + new String(msg.getBody()));
                //检查本地数据库看下是否成功
                return LocalTransactionState.COMMIT_MESSAGE;
            }
        });

        TransactionExcuterImpl transactionExcuter = new TransactionExcuterImpl();
        for (int i = 1; i <=1; i++) {
            try {
                Message msg = new Message("TopicTransaction",
                        "Transaction" + i,
                        "key",
                        ("Hello Rocket " + i).getBytes());
                SendResult sendResult = producer.sendMessageInTransaction(msg, transactionExcuter, "tq");
                System.out.println(sendResult);
            }catch (Exception e){
                e.printStackTrace();
            }
            TimeUnit.MILLISECONDS.sleep(1000);
        }

        Runtime.getRuntime().addShutdownHook(new Thread(new Runnable() {
            public void run() {
                producer.shutdown();
            }
        }));
        System.exit(0);

    }
}
